package edu.kit.aifb.ldstream.learner.tasks.mod;


import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;

import org.openrdf.model.BNode;
import org.openrdf.model.Literal;
import org.openrdf.model.Resource;
import org.openrdf.model.Statement;
import org.openrdf.model.URI;
import org.openrdf.model.Value;
import org.openrdf.query.BindingSet;
import org.openrdf.query.MalformedQueryException;
import org.openrdf.query.QueryEvaluationException;
import org.openrdf.query.QueryLanguage;
import org.openrdf.query.TupleQuery;
import org.openrdf.query.TupleQueryResult;
import org.openrdf.repository.RepositoryException;
import org.openrdf.repository.RepositoryResult;
import org.openrdf.rio.ParserConfig;
import org.openrdf.rio.RDFParser.DatatypeHandling;

import edu.kit.aifb.ldstream.env.Environment;
import edu.kit.aifb.ldstream.learner.LiteralChecker;
import edu.kit.aifb.ldstream.learner.out.n3.ValueDistribN3Converter;
import edu.kit.aifb.ldstream.learner.query.BNodeCache;
import edu.kit.aifb.ldstream.learner.query.InstanceCache;
import edu.kit.aifb.ldstream.learner.util.DescendingComparator;
import edu.kit.aifb.ldstream.learner.util.URIComparator;
import edu.kit.aifb.ldstream.stat.util.IDistributionChecker;

public class ValueDistrib extends LearnerTask{

	private boolean inferencing;
	private Set<Type> typeSet;
	private IDistributionChecker octave;

	public ValueDistrib(boolean inferencing, Set<Type> typeSet,
			IDistributionChecker octave) {
		this.inferencing = inferencing;
		this.typeSet = typeSet;
		this.octave = octave;
	}

	@Override
	public void run() {
		startup();
		setParserConfig();
		if(Environment.LOGGING){
			Environment.log.info("Started to learn  ValueDistributions.");
		}
		valueDistrib();
		if(Environment.LOGGING){
			Environment.log.info("ValueDistributions learned.");
		}
		shutdown();
	}


	private void valueDistrib(){
		ValueDistribN3Converter converter = new ValueDistribN3Converter(); 
		DistributionResult dr;
		for(URI type : this.typeSet){
			if(Environment.LOGGING){
				Environment.log.info("Collecting values for: " + type.stringValue());
			}
			InstanceCache instanceCache = new InstanceCache(type);
			Map<URI, Value> propSet = getValueProperties(instanceCache);
			Set<Entry<URI, Value>> entrySet = propSet.entrySet();


			for(Entry<URI, Value> entry : entrySet){
				// checking the "type" of the value 
				boolean discrete = true;
				boolean literal = false;
				boolean needsMapping = true;
				if(entry.getValue() instanceof Literal){
					literal = true;
					Literal lit = (Literal) entry.getValue();
					needsMapping = needsMapping(lit);

					if(LiteralChecker.isContinuousLiteral(lit.getDatatype())){
						discrete = false;
					}
				}
				//create a new cache for saving the values to a file
				ValueCache valueCache = new ValueCache();

				//collect every value of this property
				if(Environment.LOGGING){
					Environment.log.info("Collecting values for: " + type.stringValue() + " and property: " + entry.getKey().stringValue());
				}
				int numValues = collectValues(type, entry, instanceCache, valueCache, literal, discrete);
				valueCache.closeWriter();
				//compute the distribution
				if(Environment.LOGGING){
					Environment.log.info("Compute distributions for: " + type.stringValue() + " and property: " + entry.getKey().stringValue());
				}
				dr = computeValueDistribution(valueCache, numValues, discrete, needsMapping);
				//close the cache file
				valueCache.closeReader();
				valueCache = null;
				//save to file
				converter.storeToFile(type, entry.getKey(), dr);
				System.gc();
			}
		}
		converter.close();

	}

	private DistributionResult computeValueDistribution(ValueCache cache, int numValues, boolean discrete, boolean needsMapping){
		DistributionResult result = new DistributionResult();

		if(discrete){
			String[] stringValues = null;
			Integer[] intValues = null;
			List<String> mapping = null;
			Set<String> mappingSet = null;
			if(needsMapping){
				stringValues = new String[numValues];
				mapping = new LinkedList<String>();
				mappingSet = new TreeSet<String>(new DescendingComparator());
			}else{
				intValues = new Integer[numValues];
			}
			int i = 0;
			String input = cache.next();
			while(input != null){
				if(needsMapping){
					if(!mappingSet.contains(input)){
						mappingSet.add(input);
						mapping.add(input);
					}
				}
				if(needsMapping){
					stringValues[i++] = input;
				}else{
					intValues[i++] = Integer.valueOf(input);
				}
				input = cache.next();
			}
			if(Environment.LOGGING) Environment.log.info("Sending the " + numValues + " values to Octave.");
			if(needsMapping){
				result.distrib = this.octave.discrete(stringValues, needsMapping);
				result.mapping = mapping;
			}else{
				result.distrib = this.octave.discrete(intValues, needsMapping);
			}
			if(Environment.LOGGING) Environment.log.info("Result returned from Octave.");
		}else{
			double[] values = new double[numValues];
			int i = 0;
			String input = cache.next() ;
			while(input != null){
				values[i++] = Double.parseDouble(input);
				input = cache.next() ;
			}
			if(Environment.LOGGING) Environment.log.info("Sending the " + numValues + " values to Octave.");
			result.distrib = this.octave.continuous(values);
			if(Environment.LOGGING) Environment.log.info("Result returned from Octave.");
		}
		return result;
	}

	private int collectValues(URI type, Entry<URI, Value> entry, InstanceCache ic, ValueCache cache, boolean literal, boolean discrete){
		URI instance;
		int numValues = 0;
		Iterator<URI> iter = ic.iterator();

		if(iter.hasNext()){
			instance = iter.next();
			// check the first time for direct link
			boolean direct = isDirectlyConnected(instance, entry.getKey());
			if(direct){
				numValues += handleDirectLink(instance, entry, cache, literal, discrete);
			}else{
				// crawl bnode links
				numValues += handleBNodeLinks(instance, entry, cache, literal, discrete);
			}
			while(iter.hasNext()){
				instance = iter.next();
				if(direct){
					numValues += handleDirectLink(instance, entry, cache, literal, discrete);
				}else{
					// crawl bnode links
					numValues += handleBNodeLinks(instance, entry, cache, literal, discrete);
				}
			} 
		}

		return numValues;
	}

	private int handleBNodeLinks(Resource subject, Entry<URI, Value> entry, ValueCache cache, boolean literal, boolean discrete){
		int numValues = 0;

		if(isDirectlyConnected(subject, entry.getKey())){
			//we have found the direct connection save the value
			numValues += handleDirectLink(subject, entry, cache, literal, discrete);
		}else{
			// there are possible next steps (connected bnodes)
			//List<BNode> nextStep = getNextSteps(subject);
			BNodeCache bCache = getNextSteps(subject);
			for(BNode b : bCache){
				//crawl the next steps
				numValues += handleBNodeLinks(b, entry, cache, literal, discrete);
			}

			bCache.close();
			bCache = null;
		}
		return numValues;
	}

	private int handleDirectLink(Resource subject, Entry<URI, Value> entry, ValueCache cache, boolean literal, boolean discrete){
		int numValues = 0;
		// direct repository query possible
		RepositoryResult<Statement> values;
		try {
			//values = this.con.getStatements(subject, entry.getKey(), null, this.inferencing, Environment.STREAM_CONTEXT_URI);
			values = this.con.getStatements(subject, entry.getKey(), null, this.inferencing);
			try{
				Statement s;
				while(values.hasNext()){
					numValues++;
					s = values.next();
					if(literal){
						Literal lit = (Literal) s.getObject();
						if(discrete){
							cache.write(lit.stringValue());
						}else{
							cache.write(Double.toString(lit.doubleValue()));
						}
					} else{
						cache.write(s.getObject().stringValue());
					}
				}
			}finally{
				values.close();
			}
		} catch (RepositoryException e) {
			e.printStackTrace();
		}

		return numValues;
	}

	private boolean isDirectlyConnected(Resource subject, URI pred){
		boolean direct = false;
		StringBuilder sb;
		String query = null;
		if(subject instanceof BNode){
			/*
			sb = new StringBuilder("SELECT x FROM CONTEXT <");
			sb.append(Environment.STREAM_CONTEXT_URI);
			sb.append("> {_:");
			 */
			sb = new StringBuilder("SELECT x ");
			//sb.append(" FROM {");
			sb.append(" FROM {_:");
			sb.append(subject.stringValue() );
			sb.append("} <");
			sb.append(pred.stringValue());
			sb.append("> {x} LIMIT 1 ");
			query = sb.toString();

		}else{
			/*
			sb = new StringBuilder("SELECT x FROM CONTEXT <");
			sb.append(Environment.STREAM_CONTEXT_URI);
			sb.append("> {<");
			 */
			sb = new StringBuilder("SELECT x ");
			sb.append(" FROM {<");
			sb.append(subject.stringValue() );
			sb.append(">} <");
			sb.append(pred.stringValue());
			sb.append("> {x} LIMIT 1 ");
			query = sb.toString();

		}
		sb = null;
		try {
			/*
			 * There is no ask query in SERQL we simulate it by limiting the select query to 1 checking if there is an result
			 */
			TupleQuery tupleQuery = this.con.prepareTupleQuery(QueryLanguage.SERQL, query);
			tupleQuery.setIncludeInferred(this.inferencing);
			TupleQueryResult result = tupleQuery.evaluate();
			if(result.hasNext()){
				direct = true;
			}
			result.close();
		} catch (RepositoryException e) {
			e.printStackTrace();
		} catch (MalformedQueryException e) {
			e.printStackTrace();
		} catch (QueryEvaluationException e) {
			e.printStackTrace();
		}
		return direct;
	}

	private BNodeCache getNextSteps(Resource subject){
		//	List<BNode> nextStep = new LinkedList<BNode>();
		BNodeCache bCache = new BNodeCache();
		StringBuilder sb;
		//collect the linked bnodes
		String query = null;
		if(subject instanceof BNode){
			/*
			sb = new StringBuilder("SELECT DISTINCT x FROM CONTEXT <");
			sb.append(Environment.STREAM_CONTEXT_URI);
			sb.append("> {_:");
			 */
			sb = new StringBuilder("SELECT DISTINCT x ");
			//sb.append(" FROM {");
			sb.append(" FROM {_:");
			sb.append(subject.stringValue());
			sb.append("} y {x} WHERE isBNode(x)");
			query = sb.toString();
		}else{
			/*
			sb = new StringBuilder("SELECT DISTINCT x FROM CONTEXT <");
			sb.append(Environment.STREAM_CONTEXT_URI);
			sb.append("> { <");
			 */
			sb = new StringBuilder("SELECT DISTINCT x ");
			sb.append(" FROM { <");
			sb.append(subject.stringValue());
			sb.append("> } y {x} WHERE isBNode(x)");
			query = sb.toString();
		}
		try {
		//	System.out.println(query);
			TupleQuery tupleQuery = this.con.prepareTupleQuery(QueryLanguage.SERQL, query);
			tupleQuery.setIncludeInferred(this.inferencing);
			TupleQueryResult result = tupleQuery.evaluate();
			try{
				BindingSet bs;
				while(result.hasNext()){
					bs = result.next();
					bCache.write(bs.getValue("x").stringValue());
					//nextStep.add((BNode) bs.getValue("x"));
				}
			} finally{
				result.close();
			}
		} catch (RepositoryException e) {
			e.printStackTrace();
		} catch (MalformedQueryException e) {
			e.printStackTrace();
		} catch (QueryEvaluationException e) {
			e.printStackTrace();
		}
		bCache.closeWriter();
		return bCache;

	}


	private Map<URI, Value> getValueProperties(InstanceCache ic){
		Map<URI, Value> propMap = new TreeMap<URI, Value>(new URIComparator());
		for(URI instance : ic){
			crawl(instance, propMap);
		}
		return propMap;

	}

	private void crawl(Resource subject, Map<URI, Value> propMap){
		RepositoryResult<Statement> result = null;

		try {
			//	result = this.con.getStatements(subject, null, null, this.inferencing, Environment.STREAM_CONTEXT_URI);
			result = this.con.getStatements(subject, null, null, this.inferencing);
			Statement s;
			Value object;
			try{
				while(result.hasNext()){
					s = result.next();
					object = s.getObject();
					if(object instanceof BNode){
						crawl((BNode) object, propMap);
					} else {
						// should be a Literal or a URI
						if(!propMap.containsKey(s.getPredicate())){
							propMap.put(s.getPredicate(), object);
						}
					}
				}
			} finally{
				result.close();
			}
		} catch (RepositoryException e) {
			e.printStackTrace();
		}
	}

	private boolean needsMapping(Literal lit){
		boolean needsMapping = true;
		URI check = lit.getDatatype();
		for(URI type : Environment.DISCRETE_NUMBER){
			if(type.equals(check)){
				needsMapping = false;
				break;
			}
		}
		return needsMapping;
	}

	private void setParserConfig(){
		this.con.setParserConfig(new ParserConfig(true, true, true, DatatypeHandling.VERIFY));
	}

}
